package taxianalysis


import java.text.SimpleDateFormat
import java.util.Locale
import java.util.concurrent.TimeUnit

import com.esri.core.geometry.{GeometryEngine, Point, SpatialReference}
import jsonParse.{Feature, FeatureCollection, FeatureExtraction}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

import scala.io.{BufferedSource, Source}

/**
 * ClassName: TaskAnalysis <br/>
 * Description: <br/>
 * date: 2020/8/13 11:55<br/>
 *
 * @author Hesion<br/>
 * @version
 * @since JDK 1.8
 */
object TaskAnalysis {
  val spark: SparkSession = SparkSession.builder().master("local[6]").appName("Taxi").getOrCreate()
  import spark.implicits._
  import org.apache.spark.sql.functions._

  def main(args: Array[String]): Unit = {
    val taxiRow: DataFrame = spark.read
      .option("header", value = true)
      .csv("dataset/half_trip.csv")
    val taxiParsed: RDD[Either[Trip, (Row, Exception)]] = taxiRow.rdd.map(safe(parse))
    //    val taxiGood: Dataset[Trip] = taxiParsed.map(either => either.left.get).toDS()
    //上面的代码没有报错，如果报错的话，可以用下面的代码进行过滤
    val taxiGood: Dataset[Trip] = taxiParsed.filter(either => either.isLeft).map(either => either.left.get).toDS()
    val hours  = (pickupTime: Long,dropOffTime:Long)=>{
       val duration = dropOffTime-pickupTime
      val hours: Long = TimeUnit.HOURS.convert(duration, TimeUnit.MICROSECONDS)
      hours
    }
    val hoursudf: UserDefinedFunction = udf(hours)
    taxiGood.groupBy(hoursudf($"pickupTime",$"dropOffTime") as "duration").count().sort("duration").show()
    spark.udf.register("hours",hours)
    val taxiClean: Dataset[Trip] = taxiGood.where("hours(pickUpTime,dropOffTime) BETWEEN 0 AND 3")
    taxiClean.show()

    //增加行政区信息
    //7.1 读取数据集
    val geoJson: String = Source.fromFile("dataset/nyc-borough-boundaries-polygon.geojson").mkString
    val fetureCollection: FeatureCollection = FeatureExtraction.parseJson(geoJson)
    //7.2排序
    //后续需要得到每个出租车在哪个行政区。拿到经纬度。遍历features搜索其所在行政区
    //行政区越大命中几率越高，减少遍历次数
    val sortedFetures = fetureCollection.features.sortBy(feture=>{
      (feture.properties("borough"),-feture.getGeometry().calculateArea2D())
    })
    //7.3减少数据集的拷贝量，进行广播
    val feturesBC = spark.sparkContext.broadcast(sortedFetures)
    //7.4UDF创建，完成我们想要的功能
    //传入经纬度，返回行政区
    val boroughLookUp = (x:Double,y:Double)=>{

      //7.4.1命中行政区经纬度的信息
      val feturesHit: Option[Feature] = feturesBC.value.find(feture =>{
        //第一个参数传入处理后地理数据集，第二个参数传入点经纬坐标，传入默认地理参数
        GeometryEngine.contains(feture.getGeometry(),new Point(x,y),SpatialReference.create(4326))
      })
      //从命中集里取出行政区信息
      val borough = feturesHit.map(feature=>{
        feature.properties("borough")
      }).getOrElse("NA")
      borough
    }
    //7.5统计信息,获取一个行政区匹配集，待会用于和实际的数据进行匹配统计
    val boroughUDF  = udf(boroughLookUp)
    taxiClean.groupBy(boroughUDF('dropOffx,'dropOffy)).count().show()



  }

  def safe[P, R](f: P => R): P => Either[R, (P, Exception)] = {
    (param: P) => {
      try {
        Left(f(param))
      } catch {
        case e: Exception => Right((param, e))
      }
    }
  }

  /**
   * 将时间类型数据转为时间戳, 方便后续的处理
   *
   * @param row   行数据, 类型为 RichRow, 以便于处理空值
   * @param field 要处理的时间字段所在的位置
   * @return 返回 Long 型的时间戳
   */
  def parseTime(row: RichRow, field: String): Long = {
    //转换成时间的准备格式化器
    val pattern = "yyyy-MM-dd HH:mm:ss"
    val formatter = new SimpleDateFormat(pattern, Locale.ENGLISH)
    //从数据集读取出来的实际值
    val timeOption: Option[String] = row.getAs[String](field)
    //timeOption是Option类型通过map算子转换成Long类型
    timeOption.map(time => formatter.parse(time).getTime).getOrElse(0L)

  }

  def parseLocation(row: RichRow, field: String): Double = {
    row.getAs[String](field).map(loc => loc.toDouble).getOrElse(0.0D)
  }

  /**
   * step 2
   * 将Row对象转为Trip对象，从而将DataFrame转为Dataset[Trip] 方便后续操作
   *
   * @param row DataFrame中的Row对象
   * @return 代表数据集中一条记录的Trip对象
   */
  def parse(row: Row): Trip = {
    val richRow: RichRow = new RichRow(row)
    val license: String = richRow.getAs[String]("hack_license").orNull
    val pickUpTime = parseTime(richRow, "pickup_datetime")
    val dropOffTime = parseTime(richRow, "dropoff_datetime")
    val pickUpX = parseLocation(richRow, "pickup_longitude")
    val pickUpY = parseLocation(richRow, "pickup_latitude")
    val dropOffX = parseLocation(richRow, "dropoff_longitude")
    val dropOffY = parseLocation(richRow, "dropoff_latitude")

    // 创建 Trip 对象返回
    Trip(license, pickUpTime, dropOffTime, pickUpX, pickUpY, dropOffX, dropOffY)
  }
}

class RichRow(row: Row) {
  def getAs[T](field: String): Option[T] = {
    if (row.isNullAt(row.fieldIndex(field)) || StringUtils.isBlank(row.getAs[String](field))) {
      None
    } else {
      Some(row.getAs[T](field))
    }
  }
}

/**
 * step1:代表一个行程, 是集合中的一条记录
 *
 * @param license     出租车执照号
 * @param pickUpTime  上车时间
 * @param dropOffTime 下车时间
 * @param pickUpX     上车地点的经度
 * @param pickUpY     上车地点的纬度
 * @param dropOffX    下车地点的经度
 * @param dropOffY    下车地点的纬度
 */
case class Trip(
                 license: String,
                 pickUpTime: Long,
                 dropOffTime: Long,
                 pickUpX: Double,
                 pickUpY: Double,
                 dropOffX: Double,
                 dropOffY: Double
               )